package cc.xpcas.raft.network;

import java.util.concurrent.*;

import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import cc.xpcas.raft.context.Context;
import cc.xpcas.raft.context.Peer;
import cc.xpcas.raft.message.MessageContext;

/**
 * @author xp
 */
public class Poller {

    private static final Logger LOG = LoggerFactory.getLogger(Poller.class);

    private final Context context;

    private final ExecutorService executor = Executors.newCachedThreadPool();

    public Poller(Context context) {
        this.context = context;
    }

    public void start(Endpoint endpoint) {
        executor.execute(() -> poll(endpoint));
    }

    private void poll(Endpoint endpoint) {
        while (endpoint.isReceivable()) {
            Packet packet = endpoint.receive();
            if (packet == null) {
                if (LOG.isErrorEnabled()) {
                    LOG.error("receive null");
                }
                break;
            }
            Peer peer = context.getPeerByAddress(packet.getSrc());
            MessageContext messageContext = peer.wrapReceivedMessage(packet.getMessage());
            context.enqueueMessage(messageContext);
        }
    }
}
